package com.ctrip.framework.apollo.configservice.service;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.ctrip.framework.apollo.biz.config.BizConfig;
import com.ctrip.framework.apollo.biz.repository.AppNamespaceRepository;
import com.ctrip.framework.apollo.common.entity.AppNamespace;
import com.ctrip.framework.apollo.configservice.wrapper.CaseInsensitiveMapWrapper;
import com.ctrip.framework.apollo.core.ConfigConsts;
import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;
import com.ctrip.framework.apollo.core.utils.StringUtils;
import com.ctrip.framework.apollo.tracer.Tracer;
import com.ctrip.framework.apollo.tracer.spi.Transaction;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**实现InitializingBean接口，缓存AppNamespace的Service实现类。通过将AppNamespace缓存在内存中，提高查询性能
 * 启动时，全量初始化AppNamespace到缓存
 * 考虑AppNamespace新增，后台定时任务，定时增量初始化AppNamespace到缓存
 * 考虑AppNamespace更新与删除，后台定时任务，定时全量重建AppNamespace到缓存
 */
@Service
public class AppNamespaceServiceWithCache implements InitializingBean {
	private static final Logger logger = LoggerFactory.getLogger(AppNamespaceServiceWithCache.class);
	private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR).skipNulls();
	private final AppNamespaceRepository appNamespaceRepository;
	private final BizConfig bizConfig;
	// 增量初始化周期
	private int scanInterval;
	// 增量初始化周期单位
	private TimeUnit scanIntervalTimeUnit;
	// 重建周期
	private int rebuildInterval;
	// 重建周期单位
	private TimeUnit rebuildIntervalTimeUnit;
	// 定时任务ExecutorService
	private ScheduledExecutorService scheduledExecutorService;
	// 最后扫描到的AppNamespace的编号
	private long maxIdScanned;

	// 公用类型的AppNamespace的缓存
	// store namespaceName -> AppNamespace
	private CaseInsensitiveMapWrapper<AppNamespace> publicAppNamespaceCache;

	// App下的AppNamespace的缓存
	// store appId+namespaceName -> AppNamespace
	private CaseInsensitiveMapWrapper<AppNamespace> appNamespaceCache;

	// AppNamespace的缓存
	// store id -> AppNamespace
	private Map<Long, AppNamespace> appNamespaceIdCache;

	public AppNamespaceServiceWithCache(final AppNamespaceRepository appNamespaceRepository, final BizConfig bizConfig) {
		this.appNamespaceRepository = appNamespaceRepository;
		this.bizConfig = bizConfig;
		initialize();
	}

	private void initialize() {
		maxIdScanned = 0;
		// 创建缓存对象
		publicAppNamespaceCache = new CaseInsensitiveMapWrapper<>(Maps.newConcurrentMap());
		appNamespaceCache = new CaseInsensitiveMapWrapper<>(Maps.newConcurrentMap());
		appNamespaceIdCache = Maps.newConcurrentMap();
		// 创建ScheduledExecutorService对象，大小为1
		scheduledExecutorService = Executors.newScheduledThreadPool(1, ApolloThreadFactory.create("AppNamespaceServiceWithCache", true));
	}

	public AppNamespace findByAppIdAndNamespace(String appId, String namespaceName) {
		Preconditions.checkArgument(!StringUtils.isContainEmpty(appId, namespaceName), "appId and namespaceName must not be empty");
		return appNamespaceCache.get(STRING_JOINER.join(appId, namespaceName));
	}

	public List<AppNamespace> findByAppIdAndNamespaces(String appId, Set<String> namespaceNames) {
		Preconditions.checkArgument(!Strings.isNullOrEmpty(appId), "appId must not be null");
		if (namespaceNames == null || namespaceNames.isEmpty()) {
			return Collections.emptyList();
		}
		List<AppNamespace> result = Lists.newArrayList();
		for (String namespaceName : namespaceNames) {
			AppNamespace appNamespace = appNamespaceCache.get(STRING_JOINER.join(appId, namespaceName));
			if (appNamespace != null) {
				result.add(appNamespace);
			}
		}
		return result;
	}

	public AppNamespace findPublicNamespaceByName(String namespaceName) {
		Preconditions.checkArgument(!Strings.isNullOrEmpty(namespaceName), "namespaceName must not be empty");
		return publicAppNamespaceCache.get(namespaceName);
	}

	public List<AppNamespace> findPublicNamespacesByNames(Set<String> namespaceNames) {
		if (namespaceNames == null || namespaceNames.isEmpty()) {
			return Collections.emptyList();
		}

		List<AppNamespace> result = Lists.newArrayList();
		for (String namespaceName : namespaceNames) {
			AppNamespace appNamespace = publicAppNamespaceCache.get(namespaceName);
			if (appNamespace != null) {
				result.add(appNamespace);
			}
		}
		return result;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		// 从ServerConfig中，读取定时任务的周期配置
		populateDataBaseInterval();
		scanNewAppNamespaces(); // block the startup process until load finished
		scheduledExecutorService.scheduleAtFixedRate(() -> {
			Transaction transaction = Tracer.newTransaction("Apollo.AppNamespaceServiceWithCache", "rebuildCache");
			try {
				this.updateAndDeleteCache();
				transaction.setStatus(Transaction.SUCCESS);
			} catch (Throwable ex) {
				transaction.setStatus(ex);
				logger.error("Rebuild cache failed", ex);
			} finally {
				transaction.complete();
			}
		}, rebuildInterval, rebuildInterval, rebuildIntervalTimeUnit);
		scheduledExecutorService.scheduleWithFixedDelay(this::scanNewAppNamespaces, scanInterval, scanInterval, scanIntervalTimeUnit);
	}

	private void scanNewAppNamespaces() {
		Transaction transaction = Tracer.newTransaction("Apollo.AppNamespaceServiceWithCache", "scanNewAppNamespaces");
		try {
			this.loadNewAppNamespaces();
			transaction.setStatus(Transaction.SUCCESS);
		} catch (Throwable ex) {
			transaction.setStatus(ex);
			logger.error("Load new app namespaces failed", ex);
		} finally {
			transaction.complete();
		}
	}

	// for those new app namespaces
	private void loadNewAppNamespaces() {
		boolean hasMore = true;
		while (hasMore && !Thread.currentThread().isInterrupted()) {
			// 获得大于maxIdScanned的500条AppNamespace记录，按照id升序
			List<AppNamespace> appNamespaces = appNamespaceRepository.findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned);
			if (CollectionUtils.isEmpty(appNamespaces)) {
				break;
			}
			// 合并到AppNamespace缓存中
			mergeAppNamespaces(appNamespaces);
			// 获得新的maxIdScanned，取最后一条记录
			int scanned = appNamespaces.size();
			maxIdScanned = appNamespaces.get(scanned - 1).getId();
			// 若拉取不足500条，说明无新消息了
			hasMore = scanned == 500;
			logger.info("Loaded {} new app namespaces with startId {}", scanned, maxIdScanned);
		}
	}

	private void mergeAppNamespaces(List<AppNamespace> appNamespaces) {
		for (AppNamespace appNamespace : appNamespaces) {
			appNamespaceCache.put(assembleAppNamespaceKey(appNamespace), appNamespace);
			appNamespaceIdCache.put(appNamespace.getId(), appNamespace);
			if (appNamespace.isPublic()) {
				publicAppNamespaceCache.put(appNamespace.getName(), appNamespace);
			}
		}
	}

	// for those updated or deleted app namespaces
	private void updateAndDeleteCache() {
		// 从缓存中，获得所有的AppNamespace编号集合
		List<Long> ids = Lists.newArrayList(appNamespaceIdCache.keySet());
		if (CollectionUtils.isEmpty(ids)) {
			return;
		}
		// 每500一批，从数据库中查询最新的AppNamespace信息
		List<List<Long>> partitionIds = Lists.partition(ids, 500);
		for (List<Long> toRebuild : partitionIds) {
			Iterable<AppNamespace> appNamespaces = appNamespaceRepository.findAllById(toRebuild);

			if (appNamespaces == null) {
				continue;
			}

			// 处理更新的情况
			Set<Long> foundIds = handleUpdatedAppNamespaces(appNamespaces);

			// 处理删除的情况
			handleDeletedAppNamespaces(Sets.difference(Sets.newHashSet(toRebuild), foundIds));
		}
	}

	// for those updated app namespaces
	private Set<Long> handleUpdatedAppNamespaces(Iterable<AppNamespace> appNamespaces) {
		Set<Long> foundIds = Sets.newHashSet();
		for (AppNamespace appNamespace : appNamespaces) {
			foundIds.add(appNamespace.getId());
			// 获得缓存中的AppNamespace对象
			AppNamespace thatInCache = appNamespaceIdCache.get(appNamespace.getId());
			// 从DB中查询到的AppNamespace的更新时间更大，才认为是更新
			if (thatInCache != null && appNamespace.getDataChangeLastModifiedTime().after(thatInCache.getDataChangeLastModifiedTime())) {
				// 添加到appNamespaceIdCache中
				appNamespaceIdCache.put(appNamespace.getId(), appNamespace);
				String oldKey = assembleAppNamespaceKey(thatInCache);
				String newKey = assembleAppNamespaceKey(appNamespace);
				// 添加到appNamespaceCache中
				appNamespaceCache.put(newKey, appNamespace);

				// 当appId或namespaceName发生改变的情况，将老的移除出appNamespaceCache
				if (!newKey.equals(oldKey)) {
					appNamespaceCache.remove(oldKey);
				}

				if (appNamespace.isPublic()) {
					publicAppNamespaceCache.put(appNamespace.getName(), appNamespace);

					// in case namespaceName changes
					if (!appNamespace.getName().equals(thatInCache.getName()) && thatInCache.isPublic()) {
						publicAppNamespaceCache.remove(thatInCache.getName());
					}
				} else if (thatInCache.isPublic()) {
					// just in case isPublic changes
					publicAppNamespaceCache.remove(thatInCache.getName());
				}
				logger.info("Found AppNamespace changes, old: {}, new: {}", thatInCache, appNamespace);
			}
		}
		return foundIds;
	}

	// for those deleted app namespaces
	private void handleDeletedAppNamespaces(Set<Long> deletedIds) {
		if (CollectionUtils.isEmpty(deletedIds)) {
			return;
		}
		for (Long deletedId : deletedIds) {
			// 从appNamespaceIdCache中移除
			AppNamespace deleted = appNamespaceIdCache.remove(deletedId);
			if (deleted == null) {
				continue;
			}
			// 从appNamespaceCache中移除
			appNamespaceCache.remove(assembleAppNamespaceKey(deleted));
			// 从publicAppNamespaceCache移除
			if (deleted.isPublic()) {
				AppNamespace publicAppNamespace = publicAppNamespaceCache.get(deleted.getName());
				// in case there is some dirty data, e.g. public namespace deleted in some app
				// and now created in another app
				if (publicAppNamespace == deleted) {
					publicAppNamespaceCache.remove(deleted.getName());
				}
			}
			logger.info("Found AppNamespace deleted, {}", deleted);
		}
	}

	// 拼接 appId + name
	private String assembleAppNamespaceKey(AppNamespace appNamespace) {
		return STRING_JOINER.join(appNamespace.getAppId(), appNamespace.getName());
	}

	private void populateDataBaseInterval() {
		scanInterval = bizConfig.appNamespaceCacheScanInterval();
		scanIntervalTimeUnit = bizConfig.appNamespaceCacheScanIntervalTimeUnit();
		rebuildInterval = bizConfig.appNamespaceCacheRebuildInterval();
		rebuildIntervalTimeUnit = bizConfig.appNamespaceCacheRebuildIntervalTimeUnit();
	}

	// only for test use
	private void reset() throws Exception {
		scheduledExecutorService.shutdownNow();
		initialize();
		afterPropertiesSet();
	}
}
